package com.leyantech.ecp.agenx.translator.consumer;

import static com.leyantech.ecp.agenx.common.utils.SplitterUtils.DOT_SPLITTER;

import com.leyantech.chaos.kafka.KafkaConsumerProcessor;
import com.leyantech.ecp.agenx.common.handler.ActionHandler;
import com.leyantech.ecp.agenx.common.message.EnumWsFnType;
import com.leyantech.ecp.agenx.translator.filter.client.ClientMessageFilter;

import com.google.common.base.Preconditions;
import com.google.inject.Inject;
import io.vertx.core.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.List;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-02.
 */
@Slf4j
public class ClientMessageConsumer implements KafkaConsumerProcessor<String, String> {

  private final List<ActionHandler> actionHandlers;

  private final List<ClientMessageFilter> filters;

  @Inject
  public ClientMessageConsumer(List<ActionHandler> actionHandlers,
      List<ClientMessageFilter> filters) {
    this.filters = filters;
    Preconditions.checkArgument(
        CollectionUtils.isNotEmpty(actionHandlers), "actionHandlers is empty.");
    this.actionHandlers = actionHandlers;
  }

  @Override
  public void accept(ConsumerRecord<String, String> consumerRecord) {
    final String message = consumerRecord.value();

    log.debug("received message {}", message);

    final JsonObject clientMessage = new JsonObject(message);

    final String key = consumerRecord.key();
    final List<String> storeIdAndAssistant = DOT_SPLITTER.splitToList(key);

    JsonObject clientMessageBody = clientMessage.getJsonObject("body")
        .put("sid", storeIdAndAssistant.get(0))
        .put("aid", storeIdAndAssistant.get(1))
        .put("ts", consumerRecord.timestamp());

    for (ClientMessageFilter filter : filters) {
      if (!filter.test(clientMessage)) {
        return;
      }
    }

    final JsonObject clientMessageHeader = clientMessage.getJsonObject("header");

    final String fn = clientMessageHeader.getString("fn");

    if (!EnumWsFnType.contains(fn)) {
      log.warn("action {} is unimplemented", fn);
      return;
    }

    for (ActionHandler actionHandler : actionHandlers) {
      if (actionHandler.actionType().type.equals(fn)) {
        actionHandler.handle(clientMessageBody);
        return;
      }
    }

  }

}
